package cn.nicholasld.nmqs.message.mqtt;

import cn.nicholasld.nmqs.model.MqttConnection;
import cn.nicholasld.nmqs.utils.DataUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.springframework.scheduling.annotation.Async;
import javax.websocket.Session;
import java.io.IOException;
import java.util.UUID;

/**
 * @author NicholasLD
 * @createTime 2023/3/30 00:56
 */
@Async
public class MyMQTT extends Thread implements MqttCallback {
    Logger log = org.slf4j.LoggerFactory.getLogger(MyMQTT.class);


    private final String token;
    private final Session webSocketSession;

    private final String sendTopic;

    private final String receiveTopic;

    private final Integer qos;

    private MqttClient mqttClient;

    private MqttConnection mqttConnection =
            new MqttConnection(0,"1.1.1.1", "8083", "", "");

    public MyMQTT(String token, Session webSocketSession, String sendTopic, String receiveTopic, Integer qos) {
        this.token = token;
        this.webSocketSession = webSocketSession;
        this.sendTopic = sendTopic;
        this.receiveTopic = receiveTopic;
        this.qos = qos;
    }

    /**
     * 连接MQTT服务端
     */
    public void connect() throws MqttException {
        MqttConnection thisMQTT = DataUtil.getMqttConnectionByToken(token);
        if (thisMQTT == null) {
            log.error("[MQTT] 未找到token为" + token + "的MQTT连接配置");
            return;
        }
        mqttConnection = thisMQTT;

        //防止重复创建MQTTClient实例
        if (mqttClient == null){
            try {

                if (mqttConnection.getProtocol() == 1){
                    mqttClient = new MqttClient("tcp://"+mqttConnection.getServer()+":"+mqttConnection.getPort(),UUID.randomUUID().toString(),new MemoryPersistence());
                }else{
                    mqttClient = new MqttClient("ws://"+mqttConnection.getServer()+":"+mqttConnection.getPort()+"/mqtt",UUID.randomUUID().toString(),new MemoryPersistence());
                }

                //MqttDataService是接收订阅topic数据推送的回调
                mqttClient.setCallback(this);
            } catch (MqttException e) {
                log.error("[MQTT] 创建MQTTClient实例失败",e);
            }
        }
        MqttConnectOptions options = new MqttConnectOptions();
        //判断是否设置了用户名和密码
        if (!mqttConnection.getUsername().equals("") && !mqttConnection.getPassword().equals("")){
            options.setUserName(mqttConnection.getUsername());
            options.setPassword(mqttConnection.getPassword().toCharArray());
        }
        // 判断拦截状态
        if (!mqttClient.isConnected()) {
            try {
                mqttClient.connect(options);
                webSocketSession.getBasicRemote().sendText("[Websocket] 远程服务器连接成功");
                log.info("[MQTT] 连接成功！");
                try {
                    mqttClient.subscribe(receiveTopic, qos);
                } catch (MqttException e) {
                    log.error("[MQTT] 订阅失败", e);
                    mqttClient.close();
                }
            } catch (MqttException e) {
                log.error("[MQTT] 连接失败，可能是服务器连接信息有误",e);
                try {
                    webSocketSession.getBasicRemote().sendText("[Websocket] MQTT连接失败，可能是服务器连接信息有误");
                    mqttClient.close();
                } catch (IOException ex) {
                    log.error("[MQTT] WebSocket返回给前端信息失败");
                }
            } catch (IOException e) {
                log.error("[MQTT] WebSocket返回给前端信息失败");
            }
        } else {
            try {
                mqttClient.unsubscribe(receiveTopic);
                mqttClient.disconnect();
            } catch (MqttException e) {
                log.error("[MQTT] 断开连接失败", e);
            }

            try {
                mqttClient.connect(options);
                webSocketSession.getBasicRemote().sendText("[Websocket] 远程服务器连接成功");
                log.info("[MQTT] 连接成功！");
                try {
                    mqttClient.subscribe(receiveTopic, qos);
                } catch (MqttException e) {
                    log.error("[MQTT] 订阅失败", e);
                }
            } catch(MqttException e) {
                log.error("[MQTT] 连接失败，可能是服务器连接信息有误",e);
                try {
                    webSocketSession.getBasicRemote().sendText("[Websocket] MQTT连接失败，可能是服务器连接信息有误");
                    mqttClient.close();
                    this.interrupt();
                } catch (IOException ex) {
                    log.error("[MQTT] WebSocket返回给前端信息失败");
                }
            } catch (IOException e) {
                log.error("[MQTT] WebSocket返回给前端信息失败");
            }
        }


    }

    /**
     * 启动程序并订阅Topic
     * @PostConstruct 注解的作用是服务初始化启动完成后会主动执行有该注解的方法
     */
    @Async
    public void start(){
        log.info("MQTT连接数据初始化成功，开始连接....");
        try {
            connect();
        } catch (MqttException e) {
            log.error("[MQTT] 连接失败",e);
        }
    }

    /**
     * 发送数据
     */
    public void publish(String topic, String content) throws MqttException {
        MqttMessage message = new MqttMessage(content.getBytes());
        try{
            mqttClient.publish(topic, message);
        }catch (Exception e){
            log.error("[MQTT] 发送失败",e);
        }
    }

    public void publish(String content) {
        MqttMessage message = new MqttMessage(content.getBytes());
        try{
            mqttClient.publish(sendTopic, message);
        }catch (Exception e){
            log.error("[MQTT] 发送失败",e);
        }
    }

    /**
     *断线重连
     */
    public void reConnect() throws MqttException {
        if (null != mqttClient){
            MqttConnectOptions options = new MqttConnectOptions();
            mqttClient.connect(options);
            log.info("[MQTT] 重连成功！");
        }
    }

    public String getToken() {
        return token;
    }

    public Session getWebSocketSession() {
        return webSocketSession;
    }

    @Override
    public void connectionLost(Throwable throwable) {
        log.error("[MQTT] 连接断开，系统尝试重连中...",throwable);
        try {
            reConnect();
        } catch (MqttException e) {
            log.error("[MQTT] 重连失败",e);
        }
    }

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        log.info("[MQTT] 收到消息：{}",mqttMessage.toString());
        //将消息发送到前端
        webSocketSession.getBasicRemote().sendText(mqttMessage.toString());
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("[MQTT] 发送消息完成");
    }

    @Override
    public void run() {
        start();
    }

    /**
     * 释放资源
     */
    public void release() {
        try {
            mqttClient.unsubscribe(receiveTopic);
            mqttClient.disconnect();
            mqttClient.close();
            this.interrupt();
        } catch (MqttException e) {
            log.error("[MQTT] 释放资源失败", e);
        }
    }
}
